package com.atguigu.sink

import java.util

import com.atguigu.api.SensorReading
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests

/**
 * @ClassName ElasticSearchSinkTest
 * @Description
 *             查看es中的index
 *             http://localhost:9200/_cat/indices?v
 *             查看es中的数据
 *             http://localhost:9200/sensor/_search?pretty
 * @Author Mr Yang
 * @Date 2020/9/1 21:32
 * @Version 1.0
 */
object ElasticSearchSinkTest {
  def main(args: Array[String]): Unit = {
    //获取执行环境和设置并行度
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //source
    val inputStream = env.readTextFile("F:\\work\\FlinkTutorial\\src\\main\\resources\\sensor.txt")

    //transform
    val dataStream = inputStream.map(
      data => {
        val dataArray = data.split(",")
        SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
      }
    )

    //es
    val httpHosts = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("localhost", 9200))

    //创建一个es sink的builder
    val esSinkBuilder = new ElasticsearchSink.Builder[SensorReading](
      httpHosts,
      new ElasticsearchSinkFunction[SensorReading] {
        override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
          println("saving data: " + t)
          val map = new util.HashMap[String, String]()
          map.put("sensor_id", t.id)
          map.put("temperature", t.temperature.toString)
          map.put("timestamp", t.timestamp.toString)

          //创建IndexRequest，准备发送数据
          val indexRequest = Requests.indexRequest()
              .index("sensor")
              .`type`("data")
              .source(map)

          //利用indexer 发送IndexRequest
          requestIndexer.add(indexRequest)
          println("saved ...")
        }
      }
    )

    //sink
    dataStream.addSink( esSinkBuilder.build() )

    env.execute("elasticsearch sink test")
  }
}
